package app

import (
	"context"
	"fmt"
	"time"

	"code.byted.org/bytertc/whiteboard_server/biz/service/transcode/monitor"

	"github.com/google/uuid"

	"code.byted.org/bytertc/whiteboard_server/biz/pkg/grpc/grpc_client"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/grpc/grpc_server"

	"code.byted.org/bytertc/whiteboard_server/biz/pkg/rate_limit"

	"code.byted.org/bytertc/whiteboard_server/biz/pkg/object_storage"

	"code.byted.org/bytertc/whiteboard_server/biz/pkg/mq"

	"code.byted.org/bytertc/whiteboard_server/biz/pkg/gateway"
	"code.byted.org/bytertc/whiteboard_server/biz/service/replay"
	"code.byted.org/bytertc/whiteboard_server/biz/service/room"

	"code.byted.org/kite/kitex/client"

	"code.byted.org/bytertc/whiteboard_server/biz/pkg/rabbit_mq"
	"code.byted.org/bytertc/whiteboard_server/biz/service/transcode"

	"code.byted.org/bytertc/whiteboard_server/biz/config"
	"code.byted.org/bytertc/whiteboard_server/biz/dal"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/auth"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/kv/redis"
	"code.byted.org/bytertc/whiteboard_server/biz/rts"
	"code.byted.org/bytertc/whiteboard_server/kitex_gen/toutiao/videoarch/rtc_api_v2/rtcapi"
	"code.byted.org/gopkg/env"
	"code.byted.org/gopkg/logs/v2/log"
	"code.byted.org/gopkg/metrics"
	"code.byted.org/gorm/bytedgorm"
	"code.byted.org/kv/goredis"
	"code.byted.org/middleware/hertz/byted"
	"code.byted.org/middleware/hertz/pkg/app/server"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"
)

type App interface {
	Run()
}

type Whiteboard struct {
	httpServer       *server.Hertz
	tokenHelper      auth.TokenHelper
	rtsAdapter       rts.Adapter
	rtsManager       rts.Manager
	replayManager    replay.Manager
	transcodeManager transcode.Manager
	room             room.Room
	rtcAPI           gateway.RtcAPI
	repo             dal.Repo
	cfg              config.DynamicConfig
	rabbitMq         *rabbit_mq.Client
	metrics          *metrics.MetricsClientV2
	rateLimiter      *rate_limit.RateLimiter
	grpcServer       *grpc_server.GrpcServer
	transcodeMonitor *monitor.Monitor
}

type component struct {
	httpServer    *server.Hertz
	httpClient    *byted.Client
	rtcAPIClient  rtcapi.Client
	redisClient   *goredis.Client
	database      *gorm.DB
	rabbitMq      *rabbit_mq.Client
	metricsClient *metrics.MetricsClientV2
}

func (w *Whiteboard) Run() {
	go panicIfErr(w.grpcServer.Run, "grpc")

	w.registerRouter(w.httpServer)
	w.httpServer.Spin()
}

type Option func(c *component)

func New(appConfig config.AppConfig) *Whiteboard {
	log.V1.CtxInfo(context.Background(), "app config %v", appConfig)

	return newWhiteboard(
		appConfig,
		WithHTTPServer(),
		WithRTCAPIClient(appConfig.RTC),
		WithRedis(appConfig.Redis),
		WithHTTPClient(),
		WithDB(appConfig.MySQL, appConfig.DBName, appConfig.DSN),
		WithRabbitMq(appConfig.RabbitConfig),
		WithMetrics(),
	)
}

func newWhiteboard(appConfig config.AppConfig, opts ...Option) *Whiteboard {
	c := &component{}
	for _, opt := range opts {
		opt(c)
	}
	recordURL := gateway.GetURL(appConfig.Record, appConfig.RecordPath)
	transcodeURL := gateway.GetURL(appConfig.Transcode, appConfig.TranscodePath)
	rtmURL := gateway.GetURL(appConfig.RTM, appConfig.RTMPath)
	tokenURL := gateway.GetURL(appConfig.Token, appConfig.TokenPath)
	rtcAPI := gateway.NewRtcApi(c.rtcAPIClient, c.httpClient, recordURL, transcodeURL, rtmURL, tokenURL)
	cache := redis.New(c.redisClient)
	repo := dal.NewRepo(cache, c.database)
	cfg := config.NewDynamicConfig()
	m := map[string]string{appConfig.BillTopic: appConfig.BillCluster}
	kafkaSender := mq.NewKafkaSender(m)
	object_storage.NewCache()
	grpcAddr := fmt.Sprintf("/tmp/whiteboard/%s.sock", uuid.NewString())
	rtsManager := rts.NewManager(rtcAPI, grpc_client.NewConnectionManager(), c.httpClient, "unix:"+grpcAddr)
	rtsAdapter := rts.NewAdapter(repo, rtcAPI, cfg, rtsManager)
	grpcServer, err := grpc_server.NewServer(grpcAddr, rtsAdapter, rtsManager)
	if err != nil {
		panic(err)
	}

	wb := &Whiteboard{
		httpServer:       c.httpServer,
		tokenHelper:      auth.NewTokenHelper(c.rtcAPIClient),
		rtsAdapter:       rts.NewAdapter(repo, rtcAPI, cfg, rtsManager),
		rtsManager:       rtsManager,
		replayManager:    replay.NewManager(repo, cfg, rtcAPI),
		transcodeManager: transcode.NewManager(repo, cfg, c.rabbitMq, rtcAPI, c.metricsClient, appConfig.BillTopic, kafkaSender),
		room:             room.NewRoom(appConfig.BillTopic, kafkaSender, repo),
		rtcAPI:           rtcAPI,
		repo:             repo,
		cfg:              cfg,
		rabbitMq:         c.rabbitMq,
		metrics:          c.metricsClient,
		rateLimiter:      rate_limit.NewRateLimiter(c.redisClient),
		grpcServer:       grpcServer,
		transcodeMonitor: monitor.NewMonitor(repo, rtcAPI, c.metricsClient),
	}

	wb.rtsManager.StartHeartbeat()
	wb.rtsManager.StartMonitor()
	wb.transcodeMonitor.Start()

	//优雅退出
	wb.httpServer.OnShutdown = append(c.httpServer.OnShutdown, func(ctx context.Context) {
		wb.rtsManager.InstanceExit()
	})
	return wb
}

func WithHTTPServer() Option {
	return func(c *component) {
		c.httpServer = byted.Default(server.WithExitWaitTime(30 * time.Second))
	}
}

func WithHTTPClient() Option {
	httpClient, err := byted.NewClient()
	if err != nil {
		panic(err)
	}
	return func(c *component) {
		c.httpClient = httpClient
	}
}

func WithRTCAPIClient(rtcApiPSM string) Option {
	return func(c *component) {
		c.rtcAPIClient = rtcapi.MustNewClient(rtcApiPSM, client.WithCluster(env.Cluster()))
	}
}

func WithRedis(redisPSM string) Option {
	client, err := goredis.NewClient(redisPSM)
	if err != nil {
		panic(err)
	}
	return func(c *component) {
		c.redisClient = client
	}
}

func WithDB(psm string, dbName string, dsn string) Option {
	var db *gorm.DB
	var err error
	if len(dsn) != 0 {
		db, err = gorm.Open(bytedgorm.MySQL("", "").With(func(config *bytedgorm.DBConfig) {
			config.DSN = dsn
		}), &gorm.Config{
			PrepareStmt:            true,
			SkipDefaultTransaction: true,
			Logger: bytedgorm.Logger{
				LogLevel:                  logger.Info,
				IgnoreDuplicateEntryError: true,
				CompatLogger:              log.V1,
			},
			CreateBatchSize: 512,
		},
			bytedgorm.ConnPool{
				ConnMaxIdleTime: 300 * time.Second,
				ConnMaxLifetime: 300 * time.Second,
				MaxIdleConns:    100,
				MaxOpenConns:    500,
			})
	} else {
		db, err = gorm.Open(bytedgorm.MySQL(psm, dbName).WithReadReplicas(), &gorm.Config{
			SkipDefaultTransaction: true,
			Logger: bytedgorm.Logger{
				LogLevel:                  logger.Info,
				IgnoreDuplicateEntryError: true,
				CompatLogger:              log.V1,
			},
			CreateBatchSize: 30,
		},
			bytedgorm.ConnPool{
				ConnMaxIdleTime: 300 * time.Second,
				ConnMaxLifetime: 300 * time.Second,
				MaxIdleConns:    100,
				MaxOpenConns:    500,
			})
	}

	if err != nil {
		panic(err)
	}
	return func(c *component) {
		c.database = db
	}
}

func WithRabbitMq(c rabbit_mq.Config) Option {
	if env.IsBoe() {
		return func(c *component) {
			c.rabbitMq = nil
		}
	}
	rabbitMq, err := rabbit_mq.NewClient(c)
	if err != nil {
		panic(err)
	}
	return func(c *component) {
		c.rabbitMq = rabbitMq
	}
}

func WithMetrics() Option {
	cli := metrics.NewDefaultMetricsClientV2(env.PSM(), true)
	return func(c *component) {
		c.metricsClient = cli
	}
}

func panicIfErr(f func() error, hint string) {
	if err := f(); err != nil {
		panic("run " + hint + " err:" + err.Error())
	}
}
